package com.etc

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

object transformation {
  /*将原来 RDD 的每个数据项通过 map 中的用户自定义函数 f 映射转变为一个新的元素，返回类型：MappedRDD
。图 1中每个方框表示一个 RDD 分区，左侧的分区经过用户自定义函数 f:T->U 映射为右侧的新 RDD 分区。
*/
  def map(): Unit = {
    val list = List(1, 2, 3, 4, 5, 6, 7)
    list.map(a => a * 2).foreach(println)
  }
  //filter： 过滤， 不会进行shuffle
  def filter(): Unit = {
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    list.filter(a => a % 2 == 0).foreach(println)
  }
  //flatMap： 方法，可以先切分，再压平， 不会进行shuffle
  def flatMap(): Unit = {
    /*    val conf = new SparkConf().setAppName("flatMap").setMaster("local")
        val sc = new SparkContext(conf)*/
    val list = List("hello jieGe", "hello jieSao", "hello Jie")
    list.flatMap(_.split(" ")).foreach(println)

  }
/*rdd.reduceByKey(_+_)，返回值RDD[(String, Int)]，在分区内按key进行聚合
（Combiner），再把不同分区的相同的key的数据拉倒同一机器上进行聚合，会进行shuffle，
reduceByKey会进行Combiner，所以比groupByKey效率高，一般用reduceByKey
*/
  def groupByKey(): Unit = {
    val conf = new SparkConf().setAppName("groupByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(
      new Tuple2("class1", 80),
      new Tuple2("class2", 75),
      new Tuple2("class1", 90),
      new Tuple2("class2", 65))
    sc.parallelize(list).groupByKey().foreach(a => println(a._1 + "\t" + a._2))
    sc.stop()
  }
  //reduceByKey(_+_) ： 先在分区内进行按key合并，在全局合并（全局是把相同key的
  //不同的分区，拉倒同一个分区（有多个分区））会进行shuffle
  def reduceByKey(): Unit = {
    val conf = new SparkConf().setAppName("reduceByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(
      new Tuple2("class1", 80),
      new Tuple2("class2", 75),
      new Tuple2("class1", 90),
      new Tuple2("class2", 65)
    )
    sc.parallelize(list).reduceByKey(_ + _).foreach(a => println(a._1 + "\t" + a._2))
    sc.stop()
  }

/*  sortByKey ： 对于RDD内存的是元组类型的使用（进行shuffle） ，sortBy和
  sortByKey都是transformation算子，但都会触发job任务，底层new了一个rangePartition对象，底层会调用一个foreach方法，进行排序，所以会触发job
 */
  def sortByKey(): Unit = {
    val conf = new SparkConf().setAppName("sortByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(
      new Tuple2(65, "leo"),
      new Tuple2(50, "tom"),
      new Tuple2(100, "marry"),
      new Tuple2(80, "jack"))
    sc.parallelize(list).sortByKey(false).foreach(a => println(a._1 + "\t" + a._2))
    sc.stop()
  }
//rdd1.join(rdd2), 返回结果RDD[(String, (Int, Int))]，join是内连接，
//只有相同的才会join，会有shuffle过程
  def join(): Unit = {
    val conf = new SparkConf().setAppName("join").setMaster("local")
    val sc = new SparkContext(conf)
    val list1 = List(
      new Tuple2(1, "leo"),
      new Tuple2(2, "jack"),
      new Tuple2(3, "tom")
    )
    val list2 = List(
      new Tuple2(1, 100),
      new Tuple2(2, 90),
      new Tuple2(3, 60)
    )
    sc.parallelize(list1).join(sc.parallelize(list2)).foreach(a => println(a._1 + "\t" + a._2._1 + "\t" + a._2._2))
    sc.stop()
  }
/*rdd1.cogroup(rdd2) 会有shuffle，把相同的key聚合在同一机器会出现shuffle，返
回是 RDD[(String, (Iterable[Int], Iterable[Int]))]，有点类似以全外连接，所
有的rdd1和rdd2中的所有数据都能显示，String是key，第一个迭代器是对应rdd1中key的所有
value，第二个迭代器是对应rdd2中对应key的所有value，会出现如下这种结果
(jack,(CompactBuffer(3),CompactBuffer()))
(jerry,(CompactBuffer(),CompactBuffer(2，6)))
*/
  def cogroup(): Unit = {
    val conf = new SparkConf().setAppName("cogroup").setMaster("local")
    val sc = new SparkContext(conf)
    val list1 = List(
      new Tuple2(1, "leo"),
      new Tuple2(2, "jack"),
      new Tuple2(3, "tom")
    )
    val list2 = List(
      new Tuple2(1, 100),
      new Tuple2(2, 90),
      new Tuple2(3, 60),
      new Tuple2(1, 70),
      new Tuple2(2, 80),
      new Tuple2(3, 50)
    )
    sc.parallelize(list1).cogroup(sc.parallelize(list2)).foreach(a => println(a._1 + "\t" + a._2._1 + "\t" + a._2._2))
    sc.stop()
  }
/*fm.mapPartitions(it=>it.map(_*2))，传的参数是一个函数，函数的参数是迭代器，
返回是迭代器， 参数是每个分区中的数据是一个迭代器， 整个方法的返回值是 RDD[String]，
跟map类似，但map是把数据一条一条的处理，mapPartitions 是把数据一个分区一个
分区的处理,把数据放到内存。在数据量小时，推荐使用mapPartitions，在数据量大时
可能会出现内存溢出oom
*/
  def mapPartitions(): Unit = {
    val conf = new SparkConf().setAppName("mapPartitions").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    sc.parallelize(list).mapPartitions(a => a.filter(_ >= 3)).foreach(println)
    sc.stop()
  }
/*glom函数将每个分区形成一个数组，内部实现是返回的GlommedRDD。
图4中的每个方框代表一个RDD分区。图4中的方框代表一个分区。
该图表示含有V1、 V2、 V3的分区通过函数glom形成一数组Array[（V1），（V2），（V3）] */
  def glom(): Unit = {
    val conf = new SparkConf().setAppName("glom").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    sc.parallelize(list, 2).glom().foreach(println)
    sc.stop()
  }
//rdd1.union(rdd2),返回rdd1和rdd2中的所有元素，返回类型是Array，不Shuffle
  def union(): Unit = {
    val conf = new SparkConf().setAppName("union").setMaster("local")
    val sc = new SparkContext(conf)
    val list1 = List(1, 2, 3, 4, 5)
    val list2 = List(6, 7, 8, 9, 10)
     sc.parallelize(list1).union(sc.parallelize(list2)).foreach(println)
    sc.stop()
  }
  /*对 两 个 RDD 内 的 所 有 元 素 进 行 笛 卡 尔 积 操 作。
  操 作 后， 内 部 实 现 返 回CartesianRDD。
  图中左侧大方框代表两个 RDD，大方框内的小方框代表 RDD 的分区。
  右侧大方框代表合并后的 RDD，大方框内的小方框代表分区。图6中的大方框代表RDD，
  大方框中的小方框代表RDD分区
*/
  def cartesian(): Unit ={
    val conf = new SparkConf().setAppName("cartesian").setMaster("local")
    val sc = new SparkContext(conf)
    val list1 = List("A","B","C")
    val list2 = List(1, 2, 3, 4, 5)
    sc.parallelize(list1).cartesian(sc.parallelize(list2)).foreach(a => println(a._1 + "\t" + a._2))

    sc.stop()
  }
//  rdd.distinct()会发生shuffle，去重，返回类型RDD[Int]与rdd有关。
//distinct底层调用的是ReduceByKey
  def groupBy(): Unit ={
    val conf = new SparkConf().setAppName("groupBy").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val unit = sc.parallelize(list,3)
    unit.groupBy(x =>{
      if(x % 2 ==0) "even" else "odd"
    }).foreach(println)
    sc.stop()
  }
  //distinct将RDD中的元素进行去重操作。图9中的每个方框代表一个RDD分区，通过distinct函数，将数据去重
  def distinct(): Unit ={
    val conf = new SparkConf().setAppName("distinct").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1, 1, 3, 3, 5, 5, 8, 8, 9, 10)
    sc.parallelize(list).distinct().foreach(println)
    sc.stop()
  }
  /*subtract相当于进行集合的差操作，RDD 1去除RDD 1和RDD 2交集中的所有元素。
  图10中左侧的大方框代表两个RDD，大方框内的小方框代表RDD的分区。 右侧大方框
代表合并后的RDD，大方框内的小方框代表分区。
V1在两个RDD中均有，根据差集运算规则，新RDD不保留，V2在第一个RDD有，第二个RDD没有，
则在新RDD元素中包含V2
*/
  def subtract(): Unit ={
    val conf = new SparkConf().setAppName("subtract").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1,2,3,4,5,6,7,8,9)
    val list1 = List(1,3,5,7,9)
    sc.parallelize(list).subtract(sc.parallelize(list1)).foreach(println)
    sc.stop()
  }
/*sample 将 RDD 这个集合内的元素进行采样，获取所有元素的子集。
用户可以设定是否有放回的抽样、百分比、随机种子，进而决定采样方式。
内部实现是生成 SampledRDD(withReplacement， fraction， seed)。*/
  def Sample(): Unit ={
    val conf = new SparkConf().setAppName("Sample").setMaster("local")
    val sc = new SparkContext(conf)
    sc.parallelize(1 to 50).sample(false , 0.5 ,System.currentTimeMillis()
    ).foreach(println)

  }
  /*persist 函数对 RDD 进行缓存操作。
  数据缓存在哪里依据 StorageLevel 这个枚举类型进行确定。
  有以下几种类型的组合（见10）， DISK 代表磁盘，MEMORY 代表内存，
   SER 代表数据是否进行序列化存储。
下面为函数定义， StorageLevel 是枚举类型，代表存储模式，
用户可以通过按需进行选择。
persist(newLevel:StorageLevel).图 中列出persist 函数可以进行缓存的模式。
例如，MEMORY_AND_DISK_SER 代表数据可以存储在内存和磁盘，并且以序列化的方式存储，其他同理
*/
  def persist(): Unit ={
    val conf = new SparkConf().setAppName("persist").setMaster("local")
    val sc = new SparkContext(conf)
    val value = sc.parallelize(1 to 10).persist(StorageLevel.DISK_ONLY)
    val value1 = value.map(a => a * 2)
    val i = value.union(value1).reduce(_ + _)
    println(i)
    sc.stop()
  }
  //操作元祖中value ，key不改变
  def mapValues(): Unit ={
    val conf = new SparkConf().setAppName("mapValues").setMaster("local")
    val sc = new SparkContext(conf)
    sc.parallelize(Array(("a",1),("b",2),("c",3))).mapValues{_ * 10}.foreach(println)
    sc.stop()
  }

  def combineByKey(): Unit ={
    val conf = new SparkConf().setAppName("combineByKey").setMaster("local")
    val sc = new SparkContext(conf)
    val unit = sc.parallelize(Array((0,5),(0,4),(2,3),(2,2),(0,1)))
    val createCombiner = (a : Int) => List(a)
    val mergeValue = (a:List[Int],b:Int)=>a.::(b)
    val mergeCombiners=(a:List[Int],b:List[Int])=>a.:::(b)
    unit.combineByKey(createCombiner,mergeValue,mergeCombiners).foreach(println)
    sc.stop()
  }
  // rdd1.leftOuterJoin(rdd2)，返回结果是RDD[(String, (Int, Option[Int]))]，如果有rdd2中有和rdd1对应的数据时Some(value值)，没有值None，第一个Int是rdd1中的value，rdd1中的数据会显示完，有shuffle

  def leftOutJoin(): Unit ={
    val conf = new SparkConf().setAppName("leftOutJoin").setMaster("local")
    val sc = new SparkContext(conf)
    val unit1 = sc.parallelize(Array((0,1),(0,2),(2,3)))
    val unit2 = sc.parallelize(Array((0,11),(0,22),(3,33)))
    unit1.leftOuterJoin(unit2).foreach(println)
    sc.stop()
  }
  //rdd1.rightOuterJoin(rdd2)，返回结果是RDD[(String, (Option[Int], Int))]，
//  String是key类型， Option[Int]表示和rdd2中对应的没有，就和None，
//  有和rdd2中的数据对应的就是Some(value),rdd2中的数据会显示完，有shuffle
  def rightOutJoin(): Unit ={
    val conf = new SparkConf().setAppName("rightOutJoin").setMaster("local")
    val sc = new SparkContext(conf)
    val unit1 = sc.parallelize(Array((0,1),(0,2),(2,3)))
    val unit2 = sc.parallelize(Array((0,11),(0,22),(3,33)))
    unit1.rightOuterJoin(unit2).foreach(println)
    sc.stop()
  }
  //重分区partitionBy（Partitioner）
  def partitionBy(): Unit ={
    val conf = new SparkConf().setAppName("partitionBy").setMaster("local")
    val sc = new SparkContext(conf)
    sc.parallelize(Array((0,5),(0,4),(2,3),(2,2),(0,1),(1,3)))
      .partitionBy(new HashPartitioner(3))
      .foreach(println)
    sc.stop()
  }
  //元组型按value排序
  def sortBy(): Unit ={
    val conf = new SparkConf().setAppName("sortBy").setMaster("local")
    val sc = new SparkContext(conf)
    sc.parallelize(Array((0,5),(0,4),(2,3),(2,2),(0,1),(1,3)))
      .sortBy(_._2)
      .foreach(println)
    sc.stop()
  }
  //求两个集合的交集
  def intersection(): Unit ={
    val conf = new SparkConf().setAppName("intersection").setMaster("local")
    val sc = new SparkContext(conf)
    sc.parallelize(Array(1,3,4,5,8,9))
      .intersection(sc.parallelize(Array(2,4,6,7,9)))
      .foreach(println)
    sc.stop()
  }
  //分区数由多  -》 变少
  def coalesce(): Unit ={
    val conf = new SparkConf().setAppName("coalesce").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1,2,3,4,5,6,7,8,9)
    sc.parallelize(list,5).coalesce(3).foreach(println)
    sc.stop()
  }
  // 进行重分区，解决的问题：本来分区数少  -》 增加分区数
  def repartition(): Unit ={
    val conf = new SparkConf().setAppName("replication").setMaster("local")
    val sc = new SparkContext(conf)
    val list = List(1,2,3,4,5,6,7,8,9)
    sc.parallelize(list,3).repartition(5).foreach(println)
    sc.stop()
  }
  def main(args: Array[String]): Unit = {
//    map()
//    filter()
//    flatMap()
    groupByKey()
//    reduceByKey()
//    sortByKey()
//    join()
//    cogroup()
//    mapPartitions()
//    glom()
//    union()
//    cartesian()
//    groupBy()
//    distinct()
//    subtract()
//    Sample()
//    persist()
//    mapValues()
//    combineByKey()
//    leftOutJoin()
//    rightOutJoin()
//    partitionBy()
//    sortBy()
//    intersection()
//    coalesce()
//    repartition()

  }
}
